package org.dt.common.sequence;

import java.util.Arrays;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.stream.Collectors;

public class Statistic {
    /**
     * how many micro-seconds that each slot presents
     */
    private long                             msEachSlot;
    private int                              slotSize;
    //todo AtomicArrayReference may be better?
    private AtomicReference<StatisticSlot>[] statSlots;

    @SuppressWarnings("unchecked")
    Statistic(int slotSize, long msEachSlot) {
        this.statSlots  = new AtomicReference[slotSize];
        this.slotSize   = slotSize;
        this.msEachSlot = msEachSlot;

        for (int i = 0; i < slotSize; ++i) {
            statSlots[i] = new AtomicReference<>();
        }
    }

    /**
     * avg traffic that include the latest slots
     *
     * @param now         current micro-seconds
     * @param baseTraffic instead when slot is null
     * @param termFactor  the slots in top percentage are latest
     * @return avg traffic
     */
    double latestAvgTraffic(long now, long baseTraffic, float termFactor) {
        return Arrays.stream(this.statSlots).filter(statisticSlotRef -> {
            StatisticSlot slot = statisticSlotRef.get();
            return slot != null && slot.isLatestSlot(now, termFactor);
        }).map(map2TrafficValue(baseTraffic))
                     .collect(Collectors.summarizingLong(traffic -> traffic)).getAverage();
    }

    /**
     * avg traffic that does not include the latest slots
     *
     * @param now         current micro-seconds
     * @param baseTraffic instead when slot is null
     * @return avg traffic
     */
    double avgTraffic(long now, long baseTraffic) {
        return Arrays.stream(this.statSlots).filter(statisticSlotRef -> {
            StatisticSlot slot = statisticSlotRef.get();
            return slot == null || !slot.isCurrentStamp(now);
        }).map(map2TrafficValue(baseTraffic))
                     .collect(Collectors.summarizingLong(traffic -> traffic)).getAverage();
    }

    private StatisticSlot pickCurSlot(AtomicReference<StatisticSlot> curSlotRef, long now) {
        StatisticSlot pre     = curSlotRef.get();
        StatisticSlot curSlot = makeSlot(pre, now);
        while (curSlot != pre && !curSlotRef.compareAndSet(pre, curSlot)) {
            pre     = curSlotRef.get();
            curSlot = makeSlot(pre, now);
        }
        return curSlot;
    }

    public long totalTraffic(long baseTraffic) {
        return Arrays.stream(this.statSlots).map(map2TrafficValue(baseTraffic))
                     .collect(Collectors.summarizingLong(traffic -> traffic)).getSum();
    }

    void onTrafficIn(long now, long trafficCount) {
        AtomicReference<StatisticSlot> curSlotRef = getSlotRef(now);
        StatisticSlot                  curSlot    = pickCurSlot(curSlotRef, now);
        curSlot.count.addAndGet(trafficCount);
    }

    private AtomicReference<StatisticSlot> getSlotRef(long now) {
        int idx = getSlotIndex(now);
        return this.statSlots[idx];
    }

    private StatisticSlot makeSlot(StatisticSlot pre, long now) {
        if (pre == null || !pre.isCurrentStamp(now)) {
            return new StatisticSlot(now);
        } else {
            return pre;
        }
    }

    private Function<? super AtomicReference<StatisticSlot>, ? extends Long> map2TrafficValue(Long baseTraffic) {
        return statisticSlotRef -> {
            StatisticSlot slot = statisticSlotRef.get();
            if (slot == null) {
                return baseTraffic;
            } else {
                long traffic = slot.count.get();
                return Math.max(traffic, baseTraffic);
            }
        };
    }

    private long convertStamp(long now) {
        return now / msEachSlot;
    }

    private int getSlotIndex(long now) {
        return (int) (convertStamp(now) % slotSize);
    }

    private class StatisticSlot {
        private long       timestamp;
        private AtomicLong count = new AtomicLong(0L);

        StatisticSlot(long now) {
            this.timestamp = convertStamp(now);
        }

        boolean isCurrentStamp(long now) {
            return timestamp == convertStamp(now);
        }

        boolean isLatestSlot(long now, float termFactor) {
            long nowStamp    = convertStamp(now);
            long latestStamp = nowStamp - (long) (slotSize * termFactor);
            return timestamp < nowStamp && timestamp >= latestStamp;
        }

    }

}
